// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: taekwon

#pragma once

#include <experimental/filesystem>
#include <functional>
#include <memory>
#include <set>
#include <string>
#include <unordered_map>

#include <boost/algorithm/string.hpp>
#include <google/protobuf/util/message_differencer.h>
#include <unistd.h>

#include "coin/proto/coin_executor.pb.h"
#include "coin/proto/coin_order_gateway.pb.h"
#include "coin/proto/coin_request.pb.h"
#include "coin/proto/coin_strategy.pb.h"
#include "coin2/base/conversion.h"
#include "coin2/base/date_util.h"
#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/log/latency_recorder.h"
#include "coin2/exchange/base/log/proto_logger.h"
#include "coin2/exchange/base/order/order.h"

namespace coin2::exchange::base::strategy_util {

namespace {

void ValidateAccountRequest(const coin::proto::AccountRequestProto& acct_req) {
  CHECK(acct_req.has_market_type());
  CHECK(acct_req.has_exchange());
  CHECK(acct_req.has_owner());
}

};  // namespace

struct StrategyLogPathComponent {
  // e.g. lm_pass_pricer_okex_futures_allexpiry_eth.20200218-220034.1582063234889678000
  std::string strategy_name;
  std::string log_time_str;
  int64_t log_timestamp;

  const std::string GetFilenameFromLogPathComponent() const {
    return fmt::format("{}.{}.{}", strategy_name, log_time_str, log_timestamp);
  }

  static std::optional<StrategyLogPathComponent> GetLogPathComponentFromFilename(
      const std::string& filename) {
    if (std::count(filename.begin(), filename.end(), '.') != 2 &&
        std::count(filename.begin(), filename.end(), '.') != 3) {
      return {};
    }

    StrategyLogPathComponent ret;
    std::istringstream iss(filename);
    getline(iss, ret.strategy_name, '.');
    getline(iss, ret.log_time_str, '.');
    std::string log_timestamp_str;
    getline(iss, log_timestamp_str, '.');
    ret.log_timestamp = stoll(log_timestamp_str);
    return ret;
  }
};

class StrategyLogger : public BaseProtoLogger {
 public:
  using MessageDiff = google::protobuf::util::MessageDifferencer;
  using CurrencyBalance = coin::proto::CurrencyBalance;
  using ProductPosition = coin::proto::ProductPosition;
  using OrderEvent = coin::proto::OrderEvent;
  typedef std::tuple<std::string, std::string, std::string> acct_key_type;

 public:
  StrategyLogger(
      const coin::proto::StrategyRequestProto& strat_request,
      const std::string& log_root,
      const std::string& machine)
      : BaseProtoLogger(log_root, machine),
        strat_request_(strat_request),
        latency_recorder_(
            1024,
            std::bind(
                &StrategyLogger::WriteLatency,
                this,
                std::placeholders::_1,
                std::placeholders::_2,
                std::placeholders::_3)) {
    // TODO(taekwon): make this const
    process_proto_.set_machine(machine_);
    process_proto_.set_pid(static_cast<int>(getpid()));

    telemetry_.set_type(coin::proto::TelemetryProto::LATENCY);
    latency_recorder_.set_write_telemetry(std::bind(
        &StrategyLogger::WriteTelemetry,
        this,
        std::placeholders::_1,
        std::placeholders::_2,
        std::placeholders::_3));
    SetInitCallbackFunc([this](int64_t ts) { this->WriteRequest(true, ts); });
  }

  void AddCallback(std::function<void(const coin::proto::StrategyLog&, bool, int64_t)> callback) {
    callbacks_.push_back(callback);
  }

  bool
  WriteStrategyLog(const coin::proto::StrategyLog& strategy_log, bool flush, int64_t timestamp) {
    for (const auto& callback : callbacks_) {
      callback(strategy_log, flush, timestamp);
    }
    if (!strategy_log.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

  bool WriteRequest(bool flush, int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::STRAT_REQUEST);
    *strat_log_proto_.mutable_request() = strat_request_;
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WriteBalance(const coin::proto::AccountBalance& balance, bool flush, int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::BALANCE);
    *strat_log_proto_.mutable_balance() = balance;
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WriteTicker(const coin::proto::ExchangeTicker& ticker, bool flush, int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::TICKER);
    *strat_log_proto_.mutable_ticker() = ticker;
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WritePnlBalance(
      const coin::proto::CurrencyBalance& pnl_balance,
      std::optional<coin::proto::CurrencyBalance> nav_balance,
      bool flush,
      int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::PNL_BALANCE);
    *strat_log_proto_.mutable_pnl_balance() = pnl_balance;
    if (nav_balance) {
      *strat_log_proto_.mutable_nav_balance() = *nav_balance;
    }
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WritePnlBalanceDetails(
      const std::vector<AccountBalance>& pnl_bal_details,
      bool flush,
      int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::PNL_BALANCE_DETAILS);
    for (const auto& pnl_bal_detail : pnl_bal_details) {
      *(strat_log_proto_.add_pnl_balance_details()) = pnl_bal_detail;
    }
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WriteReserve(const coin::proto::AccountBalance& reserve, bool flush, int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::RESERVE);
    *strat_log_proto_.mutable_reserve() = reserve;
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WriteStratReporterConfig(
      const coin::proto::StrategyReporterConfig& config,
      bool flush,
      int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::STRAT_REPORTER_CONFIG);
    *strat_log_proto_.mutable_reporter_config() = config;
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WriteStratStoredValue(
      const coin::proto::StrategyStoredValue& value,
      bool flush,
      int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::STRAT_STORED_VALUE);
    *strat_log_proto_.mutable_strat_stored_value() = value;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WriteBalanceUpdate(
      const coin::proto::AccountBalance& balance_update,
      bool flush,
      int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::BALANCE_UPDATE);
    *strat_log_proto_.mutable_balance_update() = balance_update;
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WritePositionUpdate(
      const coin::proto::AccountPosition& position_update,
      bool flush,
      int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::POSITION_UPDATE);
    *strat_log_proto_.mutable_position_update() = position_update;
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool HasWriteBalanceCond(const coin::proto::OrderGatewayLog& order_log, int64_t curr_time) {
    if (order_log.type() != coin::proto::OrderGatewayLog::BALANCE) {
      return true;
    }
    const auto& acct_req = order_log.account_request();
    auto acct_key = std::make_tuple(
        acct_req.market_type(), acct_req.exchange(), acct_req.owner());
    if (prev_balance_send_time_.count(acct_key) == 0 ||
        prev_balance_send_time_[acct_key] + 20'000'000'000LL < curr_time) {
      prev_balance_send_time_[acct_key] = curr_time;
      return true;
    }
    return false;
  }

  bool HasWritePositionCond(const coin::proto::OrderGatewayLog& order_log, int64_t curr_time) {
    if (order_log.type() != coin::proto::OrderGatewayLog::POSITION) {
      return true;
    }
    const auto& acct_req = order_log.account_request();
    auto acct_key = std::make_tuple(
        acct_req.market_type(), acct_req.exchange(), acct_req.owner());
    if (prev_pos_send_time_.count(acct_key) == 0 ||
        prev_pos_send_time_[acct_key] + 20'000'000'000LL < curr_time) {
      prev_pos_send_time_[acct_key] = curr_time;
      return true;
    }
    return false;
  }

  void UpdateMidp(const std::string& symbol, double price) { last_midp_[symbol] = price; }

  bool WriteOrderLog(const coin::proto::OrderGatewayLog& order_log, bool flush, int64_t timestamp) {
    UpdateBalanceUpdate(order_log, flush, timestamp);
    UpdatePositionUpdate(order_log, flush, timestamp);
    // Test whether it is new balance, position
    if (!HasWriteBalanceCond(order_log, timestamp) || !HasWritePositionCond(order_log, timestamp)) {
      return false;
    }
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::OG_LOG);
    *strat_log_proto_.mutable_og_log() = order_log;
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WriteAggressiveExecutorConfig(
      const coin::proto::coin_executor::AggressiveExecutorConfig& config,
      bool flush,
      int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::AGG_EXEUCTOR_CONFIG);
    *strat_log_proto_.mutable_agg_executor_config() = config;
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WriteLatency(const coin::proto::LatencyProto& latency, bool flush, int64_t timestamp) {
    telemetry_.set_type(coin::proto::TelemetryProto::LATENCY);
    *telemetry_.mutable_latency() = latency;
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::TELEMETRY);
    *strat_log_proto_.mutable_telemetry() = telemetry_;
    *strat_log_proto_.mutable_process() = process_proto_;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WriteTelemetry(const coin::proto::TelemetryProto& telemetry, bool flush, int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::TELEMETRY);
    *strat_log_proto_.mutable_telemetry() = telemetry;
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  bool WriteExtraInfo(const std::string& extra_info, bool flush, int64_t timestamp) {
    strat_log_proto_.Clear();
    strat_log_proto_.set_type(coin::proto::StrategyLog::EXTRA_INFO);
    strat_log_proto_.set_extra_info(extra_info);
    bool ret = WriteStrategyLog(strat_log_proto_, flush, timestamp);
    return ret;
  }

  std::string GetEnsuredLogPath(int64_t timestamp) const final {
    std::string log_filename =
        StrategyLogPathComponent{strat_request_.strategy_name(), GetDateTime(timestamp), timestamp}
            .GetFilenameFromLogPathComponent();
    std::experimental::filesystem::path log_dir(log_root_);
    log_dir /= machine_;
    log_dir /= coin2::exchange::base::symbology::ToDateStr(log_timestamp());
    EnsureDirExists(log_dir.native());
    std::experimental::filesystem::path log_path(log_dir);
    log_path /= log_filename;
    return log_path.native();
  }

  void UpdateAndLogLatencyWindow(
      LatencyWindow* latency_window,
      bool flush,
      int64_t timestamp,
      const MarketExchangeApi& mea,
      coin::proto::StrategyComponentLatencyStatsProto::LatencyType lat_type,
      int64_t latency) {
    double twindow_latency_sum = 0;
    double twindow_length = 0;
    bool write = latency_window->UpdateLatencyWindow(
        timestamp,
        mea,
        lat_type,
        latency,
        &twindow_latency_sum,
        &twindow_length);
    if (write) {
      nlohmann::json obj;
      obj["tag"] = "LATENCY_INFO";
      obj["mea"] = mea.String();
      obj["latency_type"] = lat_type;
      obj["timestamp"] = timestamp;
      obj["twindow_latency_sum"] = twindow_latency_sum;
      obj["twindow_length"] = twindow_length;
      WriteExtraInfo(fmt::format("{}", obj.dump()), flush, timestamp);
    }
  }

  std::string GetLogRoot() const { return log_root_; }

  std::string GetMachine() const { return machine_; }

  coin::proto::StrategyRequestProto GetStrategyRequestProto() const { return strat_request_; }

  LatencyRecorder* mutable_latency_recorder() { return &latency_recorder_; }

  void clear_times() {
    before_feature_calc_time = std::nullopt;
    after_feature_calc_time = std::nullopt;
    before_evaluation_time = std::nullopt;
    after_evaluation_time = std::nullopt;
  }

  std::optional<int64_t> before_feature_calc_time;
  std::optional<int64_t> after_feature_calc_time;
  std::optional<int64_t> before_evaluation_time;
  std::optional<int64_t> after_evaluation_time;

 private:
  const coin::proto::StrategyRequestProto strat_request_;
  coin::proto::ProcessInfoProto process_proto_;

  coin::proto::StrategyLog strat_log_proto_;
  std::unordered_map<std::string, std::unique_ptr<CurrencyBalance>> curr_balance_;
  std::unordered_map<std::string, std::unique_ptr<ProductPosition>> curr_position_;

  std::unordered_map<std::string, double> last_midp_;

  std::string buf_;
  std::map<acct_key_type, int64_t> prev_pos_send_time_;
  std::map<acct_key_type, int64_t> prev_balance_send_time_;
  std::vector<
      std::function<void(const coin::proto::StrategyLog&, bool, int64_t)>> callbacks_;

  coin::proto::TelemetryProto telemetry_;
  LatencyRecorder latency_recorder_;

  void UpdateBalanceUpdate(
      const coin::proto::OrderGatewayLog& order_log,
      bool flush,
      int64_t timestamp) {
    if (order_log.type() != coin::proto::OrderGatewayLog::BALANCE) {
      return;
    }
    const auto& acct_req = order_log.account_request();
    ValidateAccountRequest(acct_req);
    bool has_new_update = false;
    coin::proto::AccountBalance balance_update;
    balance_update.set_market_type(acct_req.market_type());
    balance_update.set_exchange(acct_req.exchange());
    balance_update.set_owner(acct_req.owner());
    for (const auto& new_bal : order_log.balance().each_balance()) {
      auto currency = new_bal.currency();
      auto balance_key = fmt::format("{}.{}.{}.{}",
          acct_req.market_type(), acct_req.exchange(), acct_req.owner(), currency);
      if (curr_balance_.count(balance_key) == 0 ||
          !MessageDiff::ApproximatelyEquals(*curr_balance_[balance_key], new_bal)) {
        has_new_update = true;
        curr_balance_[balance_key].reset(new CurrencyBalance(new_bal));
        auto each_balance = balance_update.add_each_balance();
        *each_balance = new_bal;
      }
    }
    if (has_new_update) {
      WriteBalanceUpdate(balance_update, flush, timestamp);
    }
  }

  void UpdatePositionUpdate(
      const coin::proto::OrderGatewayLog& order_log,
      bool flush,
      int64_t timestamp) {
    if (order_log.type() != coin::proto::OrderGatewayLog::POSITION) {
      return;
    }
    const auto& acct_req = order_log.account_request();
    ValidateAccountRequest(acct_req);
    bool has_new_update = false;
    coin::proto::AccountPosition position_update;
    position_update.set_market_type(acct_req.market_type());
    position_update.set_exchange(acct_req.exchange());
    position_update.set_owner(acct_req.owner());
    for (const auto& new_pos : order_log.position().each_position()) {
      auto symbol = new_pos.symbol();
      auto position_key = fmt::format("{}.{}.{}.{}",
          acct_req.market_type(), acct_req.exchange(), acct_req.owner(), symbol);
      if (curr_position_.count(position_key) == 0 ||
          !MessageDiff::ApproximatelyEquals(*curr_position_[position_key], new_pos)) {
        has_new_update = true;
        curr_position_[position_key].reset(new ProductPosition(new_pos));
        auto each_position = position_update.add_each_position();
        *each_position = new_pos;
      }
    }
    if (has_new_update) {
      WritePositionUpdate(position_update, flush, timestamp);
    }
  }
};

class StrategySlimLogger : public StrategyLogger {
 public:
  StrategySlimLogger(
      const coin::proto::StrategyRequestProto& strat_request,
      const std::string& log_root,
      const std::string& machine)
      : StrategyLogger(strat_request, log_root, machine) {}

  bool HasWriteOglogCond(const coin::proto::OrderGatewayLog& order_log) {
    bool ret = true;
    if (order_log.type() == coin::proto::OrderGatewayLog::ORDER_EVENT) {
      CHECK(order_log.has_event());
      if (oe_type_candidates_.find(order_log.event().type()) == oe_type_candidates_.end()) {
        ret = false;
      }
    }
    return ret;
  }

  bool CondWrite(const coin::proto::StrategyLog& strategy_log, bool flush, int64_t timestamp) {
    bool has_write_cond = true;
    if (strategy_log.type() == coin::proto::StrategyLog::TELEMETRY) {
      has_write_cond = false;
    } else if (strategy_log.type() == coin::proto::StrategyLog::OG_LOG) {
      if (!HasWriteOglogCond(strategy_log.og_log())) {
        has_write_cond = false;
      }
    }
    if (!has_write_cond) {
      return false;
    }
    if (!strategy_log.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

 private:
  std::set<OrderEvent::OrderEventType> oe_type_candidates_{
      OrderEvent::ORDER_SUBMITTED,
      OrderEvent::ORDER_FILLED,
      OrderEvent::ORDER_ERROR,
      OrderEvent::ORDER_REJECTED,
      OrderEvent::CANCEL_CONFIRMED};
  std::string buf_;
};

class TelemetryLogger : public StrategyLogger {
 public:
  TelemetryLogger(
      const coin::proto::StrategyRequestProto& strat_request,
      const std::string& log_root,
      const std::string& machine)
      : StrategyLogger(strat_request, log_root, machine) {}

  bool CondWrite(const coin::proto::StrategyLog& strategy_log, bool flush, int64_t timestamp) {
    bool has_write_cond = true;
    if (strategy_log.type() == coin::proto::StrategyLog::TELEMETRY) {
      has_write_cond = true;
    } else {
      has_write_cond = false;
    }
    if (!has_write_cond) {
      return false;
    }
    if (!strategy_log.SerializeToString(&buf_)) {
      return false;
    }
    Write(buf_, flush, timestamp);
    return true;
  }

 private:
  std::string buf_;
};

class StrategyLogReader : public BaseProtoReader {
 public:
  StrategyLogReader(
      const coin::proto::StrategyRequestProto& request,
      const std::string& root_dir,
      const std::string& machine,
      int64_t begin_timestamp,
      int64_t end_timestamp,
      size_t max_record_size)
      : BaseProtoReader(root_dir, machine, begin_timestamp, end_timestamp, max_record_size),
        request_(request) {
    SetFileList(root_dir, machine, begin_timestamp, end_timestamp);
  }

  bool Read(coin::proto::StrategyLog* log_proto, int64_t* timestamp = nullptr) {
    if (timestamp) *timestamp = 0;
    ProtoLogReturn log_return;
    if (!ReadRaw(&log_return)) {
      return false;
    }
    log_proto->Clear();
    if (!log_proto->ParseFromString(log_return.log_content)) {
      return false;
    }
    if (timestamp) *timestamp = log_return.timestamp;
    return true;
  }

  virtual void SetFileList(
      const std::string& root_dir,
      const std::string& machine,
      int64_t begin_timestamp,
      int64_t end_timestamp /* exclusive */) {
    int64_t one_day = 24 * 60 * 60 * 1000000000ull;
    for (int64_t ts = begin_timestamp; ts < end_timestamp + one_day; ts += one_day) {
      std::string date_str(coin2::exchange::base::symbology::ToDateStr(ts));
      auto path = fmt::format("{}/{}/{}/", root_dir, machine, date_str);
      LOG(INFO) << "trying path " << path;
      if (!std::experimental::filesystem::exists(path)) {
        continue;
      }

      for (const auto& entry : std::experimental::filesystem::directory_iterator(path)) {
        std::string filename = entry.path().filename();
        const auto& log_path_component_optional =
            StrategyLogPathComponent::GetLogPathComponentFromFilename(filename);
        if (!log_path_component_optional) {
          LOG(WARNING) << "skip since ill-formatted filename " << filename;
          continue;
        }
        const auto& log_path_component = log_path_component_optional.value();
        tm tm_time;
        if (log_path_component.strategy_name != request_.strategy_name()) {
          LOG(WARNING) << "skip since different owner from request proto " << filename;
          continue;
        } else if (
            log_path_component.log_time_str.size() != 15 ||
            strptime(log_path_component.log_time_str.c_str(), "%Y%m%d-%H%M%S", &tm_time) ==
                nullptr) {
          LOG(WARNING) << "skip since different first log time format(yyyymmdd-hhmmss) "
                       << filename;
          continue;
        } else if (log_path_component.log_timestamp >= end_timestamp) {
          LOG(WARNING) << "skip since log timestamp is greater than end timestamp " << filename;
          continue;
        }
        files_.push_back(entry.path().string());
      }
    }
    num_files_ = files_.size();
  }

 private:
  const coin::proto::StrategyRequestProto request_;
};

}  // namespace coin2::exchange::base::strategy_util
